package com.zhentao.kafka.sample;

import lombok.SneakyThrows;
import org.apache.kafka.clients.producer.*;

import java.util.Properties;

/**
 * 生产者
 */
public class ProducerQuikStart {
    @SneakyThrows
    public static void main(String[] args) {
        //kafka的连接配置信息
        Properties prop = new Properties();
        //连接地址
        prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.233.133:9092");
        //key和value的序列表
        prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        //ack配置，消息确认机制
        prop.put(ProducerConfig.ACKS_CONFIG, "all");//all表示所有的节点都收到消息并返回的时候 1默认，表示集群首领节点收到 0表示不返回可能有消息丢失

        //重试次数
        prop.put(ProducerConfig.RETRIES_CONFIG, 10);

        //数据压缩
        prop.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");//gzip,lz4压缩的类型
        //kafka的生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(prop);
        //发布消息
        /*
        第一个参数： topic
        第二个参数： 消息的key
        第三个参数：消息的value
         */
        ProducerRecord<String, String> kvProducerRecord = new ProducerRecord<String, String>("topic-first", 0, "key-001", "hello kafka");

        //同步发送消息
        /*RecordMetadata recordMetadata = producer.send(kvProducerRecord).get();
        System.out.println(recordMetadata.offset());*/
        //异步消息发送
        producer.send(kvProducerRecord, new Callback() {
            @Override
            public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                if (e != null) {
                    System.out.println("记录异常信息到日志表中");
                }
                System.out.println(recordMetadata.offset());
            }
        });
        //关闭消息通道 必须关闭，否则消息发送不成功
        producer.close();
    }
}
